package com.binance.notification.web.push.async;

import cn.jiguang.common.ClientConfig;
import cn.jiguang.common.connection.HttpProxy;
import cn.jiguang.common.resp.ResponseWrapper;
import com.binance.notification.web.push.HttpResponseParser;
import com.binance.notification.web.push.PushResult;
import com.binance.notification.web.push.RequestCallback;

import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import org.apache.http.Consts;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.config.MessageConstraints;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.routing.HttpRoute;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.conn.NoopIOSessionStrategy;
import org.apache.http.nio.conn.SchemeIOSessionStrategy;
import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;

import java.nio.charset.CodingErrorAction;
import java.util.Arrays;

@Slf4j
public class AsyncApacheHttpClient {

    private final static Object syncLock = new Object();
    private static CloseableHttpAsyncClient _httpClient = null;
    private static OkHttpClient okClient = new OkHttpClient();
    private final int _connectionTimeout;// 设置连接超时时间，单位毫秒。
    private final int _connectionRequestTimeout;// 设置从connect
    // Manager获取Connection超时时间，单位毫秒。这个属性是新加的属性，因为目前版本是可以共享连接池的。
    private final int _socketTimeout;// 请求获取数据的超时时间，单位毫秒。 如果访问一个接口，多少时间内无法返回数据，就直接放弃此次调用
    private final int _maxRetryTimes;
    private String _authCode;
    private HttpHost _proxy;

    public AsyncApacheHttpClient(String authCode, HttpProxy proxy, ClientConfig config) {
        _maxRetryTimes = config.getMaxRetryTimes();
        _connectionTimeout = config.getConnectionTimeout();
        _connectionRequestTimeout = config.getConnectionRequestTimeout();
        _socketTimeout = config.getSocketTimeout();
        _authCode = authCode;
        if (proxy != null) {
            _proxy = new HttpHost(proxy.getHost(), proxy.getPort());
        }
    }

    private CloseableHttpAsyncClient getHttpClient(String url) throws IOReactorException {
        String hostname = url.split("/")[2];
        int port = 80;
        if (hostname.contains(":")) {
            String[] arr = hostname.split(":");
            hostname = arr[0];
            port = Integer.parseInt(arr[1]);
        }
        if (_httpClient == null) {
            synchronized (syncLock) {
                if (_httpClient == null) {
                    _httpClient = createHttpClient(10000, 10000, 10000, hostname, port);
                }
            }
        }
        return _httpClient;

    }

    public CloseableHttpAsyncClient createHttpClient(int maxTotal, int maxPerRoute, int maxRoute, String hostname,
            int port) throws IOReactorException {
        HttpHost httpHost = new HttpHost(hostname, port);
        // Create I/O reactor configuration
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime().availableProcessors()).setConnectTimeout(30000)
                .setSoTimeout(30000).setSoKeepAlive(true).setSelectInterval(2000).build();
        // Create a custom I/O reactort
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
        Registry<SchemeIOSessionStrategy> iosessionFactoryRegistry =
                RegistryBuilder.<SchemeIOSessionStrategy>create().register("http", NoopIOSessionStrategy.INSTANCE)
                        .register("https", SSLIOSessionStrategy.getDefaultStrategy()).build();
        // Create a connection manager with custom configuration.
        PoolingNHttpClientConnectionManager connManager =
                new PoolingNHttpClientConnectionManager(ioReactor, iosessionFactoryRegistry);

        // Create message constraints
        MessageConstraints messageConstraints =
                MessageConstraints.custom().setMaxHeaderCount(200).setMaxLineLength(2000).build();
        // Create connection configuration
        ConnectionConfig connectionConfig = ConnectionConfig.custom().setMalformedInputAction(CodingErrorAction.IGNORE)
                .setUnmappableInputAction(CodingErrorAction.IGNORE).setCharset(Consts.UTF_8)
                .setMessageConstraints(messageConstraints).build();
        // Configure the connection manager to use connection configuration either
        // by default or for a specific host.
        connManager.setDefaultConnectionConfig(connectionConfig);
        connManager.setConnectionConfig(httpHost, ConnectionConfig.DEFAULT);

        // Configure total max or per route limits for persistent connections
        // that can be kept in the pool or leased by the connection manager.
        connManager.setMaxTotal(maxTotal);
        connManager.setDefaultMaxPerRoute(maxPerRoute);
        connManager.setMaxPerRoute(new HttpRoute(httpHost), maxRoute);

        RequestConfig requestConfig;
        if (_proxy != null) {
            requestConfig = RequestConfig.custom().setConnectionRequestTimeout(_connectionRequestTimeout)
                    .setConnectTimeout(_connectionTimeout).setSocketTimeout(_socketTimeout).setProxy(_proxy).build();
        } else {
            requestConfig = RequestConfig.custom().setConnectionRequestTimeout(_connectionRequestTimeout)
                    .setConnectTimeout(_connectionTimeout).setSocketTimeout(_socketTimeout).build();
        }
        CloseableHttpAsyncClient client = HttpAsyncClients.custom().setConnectionManager(connManager)
                .setDefaultRequestConfig(requestConfig).build();
        client.start();
        return client;
    }

    public void sendPost(String url, String content, RequestCallback requestCallback) {
        HttpPost httpPost = new HttpPost(url);
        httpPost.setHeader(HttpHeaders.AUTHORIZATION, _authCode);
        httpPost.setHeader("Content-Type", "application/json");
        StringEntity params = new StringEntity(content, "UTF-8");
        httpPost.setEntity(params);
        try {
            getHttpClient(url).execute(httpPost, requestCallback == null ? null : new Callback(requestCallback));
        } catch (IOReactorException e) {
            if (requestCallback != null) {
                PushResult pushResult = new PushResult();
                pushResult.setEx(e);
                requestCallback.completed(Arrays.asList(pushResult));
            }
        }
    }

    private class Callback implements FutureCallback<HttpResponse> {

        private RequestCallback requestCallback;

        public Callback(RequestCallback requestCallback) {
            this.requestCallback = requestCallback;
        }

        @Override
        public void completed(HttpResponse response) {
            ResponseWrapper wrapper = new ResponseWrapper();
            HttpResponseParser.parser(response, wrapper, null);
            PushResult pushResult = PushResult.fromResponse(wrapper);
            this.requestCallback.completed(Arrays.asList(pushResult));
        }

        @Override
        public void failed(Exception ex) {
            PushResult pushResult = new PushResult();
            pushResult.setEx(ex);
            this.requestCallback.completed(Arrays.asList(pushResult));
        }

        @Override
        public void cancelled() {
            PushResult pushResult = new PushResult();
            pushResult.setEx(new Exception("请求被取消"));
            this.requestCallback.completed(Arrays.asList(pushResult));
        }

    }
}
